#!/bin/bash
# ********************************************************************************
# * Filename     : datax_push_mysql.sh
# * Author       : duanchengxin
# * Version      : 1.0.0
# * Created Date : 20241106
# * Description  : datax推mysql-DB
# ********************************************************************************

# Module 1: 设置环境变量

# Module 2: 公共函数
## 定义帮助文档
function ShowUsage() {
    cat <<EOF
USAGE: $0 [OPTIONS]

OPTIONS:
    -h --help           查看帮助文档.
    -d --dc-trade-date  [*]业务日期,格式为YYYYMMDD.
    -j --json-file-name [*]Datax Json 配置文件名.
    -p --profile        TDH环境参数,默认为生产环境.参数值可填prod、dev、backup.其中prod为生产环境、dev为开发环境、backup为灾备环境.
    --src-name-service  源 HDFS HA NameService,格式为"TDHHOST1:PORT;TDHHOST2:PORT".
    --src-jdbc          源数据库连接信息,格式为jdbc连接串.
    --src-username      源数据库用户名.
    --src-password      源数据库用户密码.
    --tar-jdbc          目标数据库连接信息,格式为jdbc连接串.
    --tar-username      目标数据库用户名.
    --tar-password      目标数据库用户密码.
    --push-days         数据推送周期,格式为"days,0/1",days:天数(正整数>=1与负整数<=-1),0:自然日,1:交易日,即推送n个自然日或交易日的数据. 
    --jvm-buffer        JVM堆参数,即为防止jvm报内存溢出,根据实际情况调整jvm的堆参数,建议将内存设置为4G或者8G.
    --inc-value         非时间戳增量标识字段值.
    --dependence        当前依赖项,格式为"table_name1;table_name2;table_name3".
    --tar-es-url        推送ES的URL信息.
    --custom-partition  自定义分区值,即非日分区传入.
    --nonpath-err       HDFS路径不存在时强制成功,该参数表示中止执行并不提示错误,否则默认将置为报错状态.

EOF
}
## 定义日志记录函数
function Logger() {
    echo -e "$(date +%Y/%m/%d\ %H:%M:%S)" : "${0##.*/}" "[${1}]"- "${2}"
}
PROFILE='prod'

# Module 3: 获取传入参数与参数值校验
## 使用 getopt 获取参数
OPTINOS=$(getopt --option hd:j:p: --long help,dc-trade-date:,json-file-name:,profile:,src-name-service:,src-jdbc:,src-username:,src-password:,tar-jdbc:,tar-username:,tar-password:,push-days:,jvm-buffer:,inc-value:,dependence:,tar-es-url:,custom-partition:,nonpath-err -n "$0" -- "$@")
VALID_OPTINOS=$?
[[ "${VALID_OPTINOS}" -ne "0" ]] && {
    echo "Try '$0 -h or --help' for more information."
    exit 10
}
## 使用 set 将 ${OPTINOS} 设置为位置参数
eval set -- "${OPTINOS}"
## 使用 shift 和 while 循环解析位置参数
while true; do
    case "$1" in
    -h | --help)
        ShowUsage
        exit 0
        ;;
    -d | --dc-trade-date)
        Logger "INFO" "业务日期: $2"
        DC_TRADE_DATE=$2
        shift 2
        ;;
    -j | --json-file-name)
        Logger "INFO" "Datax Json 配置文件名: $2"
        JSON_FILE_NAME=$2
        shift 2
        ;;
    -p | --profile)
        Logger "INFO" "profile TDH环境配置: $2"
        PROFILE=$2
        shift 2
        ;;
    --src-name-service)
        Logger "INFO" "源 HDFS HA NameService: $2"
        SRC_NAME_SERVICE=$2
        shift 2
        ;;
    --src-jdbc)
        Logger "INFO" "源数据库连接信息: $2"
        SRC_JDBC=$2
        shift 2
        ;;
    --src-username)
        Logger "INFO" "源数据库用户名: $2"
        SRC_USERNAME=$2
        shift 2
        ;;
    --src-password)
        Logger "INFO" "源数据库用户密码: $2"
        SRC_PASSWORD=$2
        shift 2
        ;;
    --tar-jdbc)
        Logger "INFO" "目标数据库连接信息: $2"
        TAR_JDBC=$2
        shift 2
        ;;
    --tar-username)
        Logger "INFO" "目标数据库用户名: $2"
        TAR_USERNAME=$2
        shift 2
        ;;
    --tar-password)
        Logger "INFO" "目标数据库用户密码: $2"
        TAR_PASSWORD=$2
        shift 2
        ;;
    --push-days)
        Logger "INFO" "数据推送周期: $2"
        PUSH_DAYS=$2
        shift 2
        ;;
    --jvm-buffer)
        Logger "INFO" "JVM堆参数: $2"
        JVM_BUFFER=$2
        shift 2
        ;;
    --inc-value)
        Logger "INFO" "非时间戳增量标识字段值: $2"
        INC_VALUE=$2
        shift 2
        ;;
    --dependence)
        Logger "INFO" "当前依赖项: $2"
        DEPENDENCE=$2
        shift 2
        ;;
    --tar-es-url)
        Logger "INFO" "推送ES的URL信息: $2"
        TAR_ES_URL=$2
        shift 2
        ;;
    --custom-partition)
        Logger "INFO" "自定义分区值: $2"
        CUSTOM_PARTITION=$2
        shift 2
        ;;
    --nonpath-err)
        Logger "INFO" "HDFS路径不存在时强制成功: True"
        NONPATH_ERR="True"
        shift 1
        ;;
    --)
        shift
        break
        ;;
    *)
        Logger "WARNING" "无效选项: $1"
        ShowUsage
        exit 10
        ;;
    esac
done
## 校验参数值
if [[ -z "${DC_TRADE_DATE}" ]]; then
    Logger "ERROR" "[-d | --dc-trade-date] : 业务日期不能为空!"
    ShowUsage
    exit 10
elif ! [[ "$(expr "${DC_TRADE_DATE}" : "[1-3][0-9][0-9][0-9][0-1][0-9][0-3][0-9]")" = "8" ]]; then
    Logger "ERROR" "[-d | --dc-trade-date] : 业务日期格式错误!"
    ShowUsage
    exit 10
elif [[ -z "${JSON_FILE_NAME}" ]]; then
    Logger "ERROR" "[-j | --json-file-name] : DATAX JSON 配置文件名不能为空!"
    ShowUsage
    exit 10
elif [[ "${JSON_FILE_NAME:(-3)}" != "add" && "${JSON_FILE_NAME:(-3)}" != "all" ]]; then
    Logger "ERROR" "[-j | --json-file-name] : 无法根据 DATAX JSON 配置文件名确认增全量推送方式,请检查是否以add/all结尾!"
    ShowUsage
    exit 10
elif [[ -n "${PUSH_DAYS}" && ! "${PUSH_DAYS}" =~ ^(-)?[1-9]+,(0|1)$ ]]; then
    Logger "ERROR" "[--push-days] : 数据推送周期格式错误!"
    ShowUsage
    exit 10
fi

# Module 4: 定义全局变量
# 日志表
LOG_TABLE_NAME="dwd:dwd_dc_etl_exec_logs"
## 记录开始时间
ETL_START_TIME="$(date "+%Y-%m-%d %H:%M:%S")"
## 日志表 任务名称
PROC_NAME=${JSON_FILE_NAME#*.}
## 日志表 任务类型,1:存储过程,2:数据推送,3:数据推送
PROC_TYPE=3
## 日志表 执行表名或者工作流名
TABLE_NAME=${JSON_FILE_NAME#*.}
## 日志表 任务执行结果标志位.1:成功,2:失败
EXEC_FLAG=1
# 是否写日志标志,0-不写入,非0-写入
IS_WRITE_LOG=1
# Inceptor 生产集群连接信息
if [[ "${PROFILE}" = "prod" ]]; then
    Logger "INFO" "TDH环境设置为生产环境"
    export HADOOP_USER_NAME=swscdc
    source /usr/local/bigdata_client/TDH-oc-Client/init.sh
    INCEPTOR_JDBC_URI="jdbc:inceptor2://10.151.6.28:10000"
    INCEPTOR_USER="swscdc"
    INCEPTOR_PASSWORD=$(echo "U3dzYzYwMDM2OTAx" | base64 -d)
elif [[ "${PROFILE}" = "dev" ]]; then
    Logger "INFO" "TDH环境设置为开发环境"
    export HADOOP_USER_NAME=swscdc
    source /usr/local/bigdata_client/TDH-dev-Client/init.sh
    INCEPTOR_JDBC_URI="jdbc:inceptor2://10.151.4.112:10000"
    INCEPTOR_USER="swscdc"
    INCEPTOR_PASSWORD=$(echo "c3dzYzYwMDM2OQo=" | base64 -d)
elif [[ "${PROFILE}" = "backup" ]]; then
    Logger "INFO" "TDH环境设置为灾备环境"
    export HADOOP_USER_NAME=swscdc
    source /usr/local/bigdata_client/TDH-lt-Client/init.sh
    INCEPTOR_JDBC_URI="jdbc:inceptor2://10.152.6.28:10000"
    INCEPTOR_USER="swscdc"
    INCEPTOR_PASSWORD=$(echo "U3dzYzYwMDM2OTAx" | base64 -d)
else
    Logger "ERROR" "PROFILE参数值错误,参数值可填prod、dev、backup.其中prod为生产环境、dev为开发环境、backup为灾备环境"
    ShowUsage
    exit 10
fi
# 设置全局DataX执行参数
PARAMS="-DparaReaderJdbc=${SRC_JDBC} -DsrcUsername=${SRC_USERNAME} -DsrcPassword=${SRC_PASSWORD} -DcustomPartition=${CUSTOM_PARTITION} -DtarEsUrl=${TAR_ES_URL} -DparaWriterJdbc=${TAR_JDBC} -DtarUsername=${TAR_USERNAME} -DtarPassword=${TAR_PASSWORD} -DincValue='${INC_VALUE}'"

# Module 5: 业务处理函数
## 获取 json 文件引入资源函数
function GetJsonFile() {
    for file in "$1"/*; do
        if test -f "$file"; then
            if [[ "${file##*/}" == "${JSON_FILE_NAME#*.}.json" ]]; then
                JSON_FILE="${file}"
            fi
        else
            GetJsonFile "${file}"
        fi
    done
}
## 调用 datax 执行函数
function PushData() {
    Logger "INFO" "[PushData] - 开始数据推送 ..."
    # 设置DataX执行参数
    local now_params
    # 判断源数据库类型,inceptor或者关系型数据库
    if [[ -n "${SRC_NAME_SERVICE}" ]]; then
        ## 推送前检查HDFS路径是否存在
        PathIsExist
        if [[ ${NOW_STATUS} == "False" ]]; then
            return
        fi
        ## HDFS HA NameService 参数值分割处理
        local src_name_service1 src_name_service2
        src_name_service1=$(echo "${SRC_NAME_SERVICE}" | awk -F ";" '{print $1}')
        src_name_service2=$(echo "${SRC_NAME_SERVICE}" | awk -F ";" '{print $2}')
        now_params="-DinTradeDate=${PARAM_TRADE_DATE} ${PARAMS} -DsrcNameService1=${src_name_service1} -DsrcNameService2=${src_name_service2}"
    else
        now_params="-DinTradeDate=${PARAM_TRADE_DATE} ${PARAMS}"
    fi
    # 根据是否设置了JVM_BUFFER参数来决定是否需要修改JVM堆参数
    if [[ -n "${JVM_BUFFER}" ]]; then
        Logger "INFO" "[PushData] - JVM堆参数 : ${JVM_BUFFER}G"
        JVM="-Xms${JVM_BUFFER}G -Xmx${JVM_BUFFER}G"
    else
        # JVM堆参数默认为2G
        JVM="-Xms2G -Xmx2G"
    fi
    # 执行DataX任务
    Logger "INFO" "[PushData] - DataX执行参数: ${now_params}"
    python /usr/local/datax/bin/datax.py --jvm="${JVM}" "${JSON_FILE}" -p"${now_params}"
    local datax_result=$?
    if [[ "${datax_result}" == 0 ]]; then
        Logger "INFO" "[PushData] - 数据推送完成."
    else
        Logger "ERROR" "[PushData] - 数据推送失败."
        exit 10
    fi
}
## 写入 Hyperbase 日志函数
function WriteEtlLogToHyperbase() {
    local key=${JSON_FILE_NAME#*.}-${PARAM_TRADE_DATE}
    local data_date=${PARAM_TRADE_DATE}
    local etl_end_time
    etl_end_time="$(date "+%Y-%m-%d %H:%M:%S")"
    local write_log_times=1
    Logger "INFO" "[WriteEtlLogToHyperbase] - 写入推送完成日志信息 ..."
    Logger "INFO" "[WriteEtlLogToHyperbase] - TABLE : [dwd.dwd_dc_etl_exec_logs]"
    Logger "INFO" "[WriteEtlLogToHyperbase] - Hyperbase写入值: {key : ${key}, proc_name : ${PROC_NAME}, proc_type : ${PROC_TYPE}, table_name : ${TABLE_NAME}, data_date : ${data_date}, etl_start_time : ${data_date}, etl_end_time : ${etl_end_time}, exec_flag : ${EXEC_FLAG}}"
    while true; do
        echo "hput '${LOG_TABLE_NAME}','${key}','f:q1','${PROC_NAME}'
              hput '${LOG_TABLE_NAME}','${key}','f:q2','${PROC_TYPE}'
              hput '${LOG_TABLE_NAME}','${key}','f:q3','${TABLE_NAME}'
              hput '${LOG_TABLE_NAME}','${key}','f:q4','${data_date}'
              hput '${LOG_TABLE_NAME}','${key}','f:q5','${ETL_START_TIME}'
              hput '${LOG_TABLE_NAME}','${key}','f:q6','${etl_end_time}'
              hput '${LOG_TABLE_NAME}','${key}','f:q7','${EXEC_FLAG}'" | hbase shell -n >/dev/null 2>&1
        local write_result=$?
        if [[ "${write_result}" == "0" ]]; then
            Logger "INFO" "[WriteEtlLogToHyperbase] - 日志信息写入完成."
            break
        else
            Logger "INFO" "[WriteEtlLogToHyperbase] - 日志信息写入失败,将进行失败重做."
            if [[ ${write_log_times} -ge 3 ]]; then
                Logger "ERROR" "[WriteEtlLogToHyperbase] - 失败重做次数已超过限制[>=3],退出执行."
                exit 10
            else
                Logger "INFO" "[WriteEtlLogToHyperbase] - 失败重做次数: [${write_log_times}], 失败重做间隔: [10s]"
                sleep 10s
                write_log_times=$((write_log_times + 1))
            fi
        fi
    done
}
## 检查当前依赖项是否完成函数
function CheckSourceByHbaseShell() {
    Logger "INFO" "[CheckSourceByHbaseShell] - 检查当前依赖项是否完成 ..."
    # 设置分隔符为分号
    IFS=';'
    # 将字符串转换为数组
    read -ra arr <<< "${DEPENDENCE}"
    # 将IFS变量重置为默认值
    IFS=$' \t\n'
    # 循环读取数组
    for elem in "${arr[@]}"; do
        local key="${elem}-${PARAM_TRADE_DATE}"
        Logger "INFO" "[CheckSourceByHbaseShell] - 依赖项: ${elem}, key: ${key}"
        while true; do
            local hbase_result check_result
            hbase_result=$(echo "hget '${LOG_TABLE_NAME}','${key}'" | hbase shell -n 2>&1)
            check_result=$(echo "${hbase_result}"| awk '/row\(s\)/{print $0}' | awk -F " " '{print $1}' | sed ':a;N;$!ba;s/\n//g' | sed s/[[:space:]]//g)
           if [[ -n "${check_result}" && "${check_result}" != "0" ]]; then
                Logger "INFO" "[CheckSourceByHbaseShell] - 检查到依赖项已完成."
                break
           else
                Logger "INFO" "[CheckSourceByHbaseShell] - 检查到依赖项未完成,将等待1m后再次检查."
                sleep 1m
           fi
        done
    done
}
## 获取自然日或交易日函数
function GetNatureOrTradeDays() {
    local is_trade_date
    DAYS=$(echo "${PUSH_DAYS}" | awk -F "," '{print $1}')
    is_trade_date=$(echo "${PUSH_DAYS}" | awk -F "," '{print $2}')
    local sub_query_sql
    local days_abs
    if [[ "${is_trade_date}" == "0" ]]; then
        DAYS_FLAG="自然日"
        if [[ "${DAYS}" -gt 0 ]]; then
            sub_query_sql="AND T.DATE >= ${DC_TRADE_DATE} ORDER BY T.DATE LIMIT ${DAYS};"
        else
            days_abs=$((0 - "${DAYS}"))
            sub_query_sql="AND T.DATE <= ${DC_TRADE_DATE} ORDER BY T.DATE DESC LIMIT ${days_abs};"
        fi
    else
        DAYS_FLAG="交易日"
        if [[ "${DAYS}" -gt 0 ]]; then
            sub_query_sql="AND T.IS_TRADE_DATE = 1 AND T.DATE >= ${DC_TRADE_DATE} ORDER BY T.DATE LIMIT ${DAYS};"
        else
            days_abs=$((0 - "${DAYS}"))
            sub_query_sql="AND T.IS_TRADE_DATE = 1 AND T.DATE <= ${DC_TRADE_DATE} ORDER BY T.DATE DESC LIMIT ${days_abs};"
        fi
    fi
    # 初始化空数组
    DAYS_ARR=()
    local query_sql="SELECT T.DATE FROM DIM.DIM_TRADE_DATE_ADD_YEAR T WHERE T.MKT_CODE = '1' ${sub_query_sql}"
    local query_result trade_days
    query_result=$(beeline -u "${INCEPTOR_JDBC_URI}" -n "${INCEPTOR_USER}" -p "${INCEPTOR_PASSWORD}" -e "${query_sql}" --showHeader=false --silent=true)
    trade_days=$(echo "${query_result}" | awk -F "|" '{print $2}' | sed ':a;N;$!ba;s/\n/;/g' | sed s/[[:space:]]//g | grep -oP '\d{8}' | paste -sd ' ' -)
    # 设置分隔符为分号
    IFS=' '
    # 将字符串转换为数组
    read -ra DAYS_ARR <<< "${trade_days}"
    # 将IFS变量重置为默认值
    IFS=$' \t\n'
}
## HDFS路径检查函数
function PathIsExist() {
    Logger "INFO" "[PathIsExist] - 开始检查HDFS路径是否存在 ..."
    local hdfs_path
    hdfs_path=$(sed -n '/.*"path": "\(.*\)",/{s//\1/p;q}' "${JSON_FILE}" | sed "s/\${inTradeDate}/${PARAM_TRADE_DATE}/g" | sed "s/\${customPartition}/${CUSTOM_PARTITION}/g")
    hdfs_path=$(echo -e "${hdfs_path}" | sed -e 's/\n//g' -e 's/\r//g')
    Logger "INFO" "[PathIsExist] - fs.defaultFS path : ${hdfs_path}"
    Logger "INFO" "[PathIsExist] - 路径检查命令 : hadoop fs -test -e ${hdfs_path} >/dev/null 2>&1"
    hadoop fs -test -e "${hdfs_path}" >/dev/null 2>&1
    local test_result=$?
    if [[ "${test_result}" == 0 ]]; then
        Logger "INFO" "[PathIsExist] - 已检查到配置项fs.defaultFS,path存在."
        Logger "INFO" "[PathIsExist] - HDFS路径检查完成."
        NOW_STATUS="True"
    else
        if [[ "${NONPATH_ERR}" == "True" ]]; then
            Logger "INFO" "[PathIsExist] - 已检查到配置项fs.defaultFS,path不存在,则默认本次无任何数据进行推送,中止执行,任务状态将置为成功！"
            Logger "INFO" "[PathIsExist] - HDFS路径检查完成."
            Logger "INFO" "[PushData] - 数据推送完成."
            NOW_STATUS="False"
        else
            Logger "ERROR" "[PathIsExist] - 无法读取路径${hdfs_path}下的所有文件,请确认您的配置项fs.defaultFS,path的值是否正确,是否有读写权限,网络是否已断开！"
            Logger "INFO" "[PathIsExist] - HDFS路径检查完成."
            Logger "ERROR" "[PushData] - 数据推送失败."
            exit 10
        fi
    fi
}
## 构建 main 函数
function Main() {
    GetJsonFile .
    ## 判断 json 文件是否是否存在且是否为空文件
    if [[ ! -s ${JSON_FILE} ]]; then
        Logger "INFO" "[Main] - 配置文件 : ${JSON_FILE_NAME#*.}.json"
        Logger "ERROR" "[Main] - 配置文件不存或者为空文件,请检查是否引入资源或者引入错误资源."
        exit 10
    fi
    ## 判断是否需要推送某个时间段内的数据
    if [[ -z "${PUSH_DAYS}" ]]; then
        ## 设置后续函数的中PARAM_TRADE_DATE值为DC_TRADE_DATE
        PARAM_TRADE_DATE="${DC_TRADE_DATE}"
        # 判断是否需要进行当前依赖项检查
        if [[ -n "${DEPENDENCE}" ]]; then
            CheckSourceByHbaseShell
        fi
        ## 推送数据
        PushData
        # 写入推送成功日志
        if [[ ${IS_WRITE_LOG} -gt 0 ]]; then
            ## 向生产环境日志表写入日志,因为旧集群未启用HBase,所以暂时不写日志.
            WriteEtlLogToHyperbase
        fi
    else
        ## 进行循环推数
        GetNatureOrTradeDays
        Logger "INFO" "[Main] - 推送近${DAYS}个${DAYS_FLAG}内的数据 ..."
        Logger "INFO" "[Main] - ${DAYS_FLAG} : ${DAYS_ARR[*]}"
        for idx in "${!DAYS_ARR[@]}"; do
            PARAM_TRADE_DATE="${DAYS_ARR[idx]}"
            Logger "INFO" "[Main] - [${idx}] - 业务日期 : ${DC_TRADE_DATE}, 数据日期 : ${PARAM_TRADE_DATE}"
            # 判断是否需要进行当前依赖项检查
            if [[ -n "${DEPENDENCE}" ]]; then
                CheckSourceByHbaseShell
            fi
            ## 推送数据
            PushData
            ## 写入推送成功日志
            if [[ ${IS_WRITE_LOG} -gt 0 ]]; then
                ## 向生产环境日志表写入日志,因为旧集群未启用HBase,所以暂时不写日志.
                WriteEtlLogToHyperbase
            fi
        done
    fi
}

# Module 6: 执行 main 函数
Main
